package org.sxp.common.utils.datax.reader;


import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Maps;
import org.sxp.common.dto.datax.reader.DataXHbaseReader;
import org.sxp.common.dto.datax.reader.DataXHiveReader;
import org.sxp.common.exception.BusinessException;
import org.sxp.common.utils.SpringContextUtils;
import org.sxp.modules.datasource.entity.DatabaseDO;
import org.sxp.modules.datasource.service.DatabaseService;
import org.sxp.modules.dataworks.dto.SyncDTO;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;

/**
 * mysql reader 构建类
 *
 * @author zhouhongfa@gz-yibo.com
 * @ClassName MysqlReader
 * @Version 1.0
 * @since 2019/7/30 23:07
 */
public class HbaseReader extends BaseReader implements IDataxReader {

    public static final String VERSION_20XSQL = "20xsql";
    public static final String VERSION_11XSQL = "11xsql";
    public static final String VERSION_11X = "11x";

    public HbaseReader(SyncDTO syncDTO) {
        super(syncDTO);
    }

    @Override
    public String getName() {
        return "hbase" + syncDTO.getSourceDbVersion() + "reader";
    }

    @Override
    public Map<String, Object> build() {
        DatabaseService databaseService = SpringContextUtils.getBean(DatabaseService.class);
        DatabaseDO jobDatasource = databaseService.getById(syncDTO.getSourceDb());
        Map<String, Object> readerObj = Maps.newLinkedHashMap();
        readerObj.put("name", getName());
        Map<String, Object> parameterObj = Maps.newLinkedHashMap();

        if(syncDTO.getSourceDbVersion().equals(VERSION_20XSQL)){

        }else if(syncDTO.getSourceDbVersion().equals(VERSION_11X)){
            version11X(parameterObj, jobDatasource);
        }else{
            throw new BusinessException("不支持的Hbase版本:"+syncDTO.getSourceDbVersion());
        }

        readerObj.put("parameter", parameterObj);
        return readerObj;
    }

    @Override
    public void backBuild() {
        String removeSpace = StrUtil.trim(syncDTO.getDataxJson());

        Matcher matcher = pattern.matcher(removeSpace);
        String value = null;
        if (matcher.find()) {
            value = matcher.group(1);
        }
        if(JSONUtil.isJson(value)){
            DataXHbaseReader reader = JSON.parseObject(value, DataXHbaseReader.class);
            if(ArrayUtil.isNotEmpty(reader.getParameter().getMode())){
                syncDTO.setSourceMode(reader.getParameter().getMode());
            }
        }
    }

    private void version11X(Map<String, Object> parameterObj, DatabaseDO jobDatasource) {
        parameterObj.put("table", syncDTO.getSourceTable());
        parameterObj.put("mode", syncDTO.getSourceMode());
        parameterObj.put("column", syncDTO.getSourceField());
        parameterObj.put("encoding", syncDTO.getSourceEncoding());

        JSONObject jsonObject = JSON.parseObject(syncDTO.getTargetHbaseConfig());
        if(jsonObject == null){
            jsonObject = new JSONObject();
        }
        jsonObject.put("hbase.zookeeper.quorum", jobDatasource.getUrl());
        parameterObj.put("hbaseConfig", jsonObject);
    }
}
